package e3dcs

import (
	"fmt"
	"runtime"
	"sync"
	"time"

	"gitee.com/ameise84/e3dcs/internal/discov"
	"gitee.com/ameise84/e3dcs/internal/log"
	"gitee.com/ameise84/e3dcs/internal/message"
	"gitee.com/ameise84/e3lock"
	"gitee.com/ameise84/e3pool/go_pool"
	"github.com/pkg/errors"
)

var _gRouterProvider routerProvider

func init() {
	_gRouterProvider = routerProvider{}
	_gRouterProvider.retryRunner = go_pool.NewGoRunner(&_gRouterProvider, "retry runner")
}

type routerProvider struct {
	localService sync.Map //本地服务
	routerMap    sync.Map
	mu           e3lock.SpinLock
	retryRunner  go_pool.GoRunner
	isRunning    bool
}

func (ts *routerProvider) E3LogMarshall() string {
	return fmt.Sprintf("router provider")
}

func (ts *routerProvider) OnPanic(err error) {
	log.GetLogger().Error("OnPanic").Object(ts).Err(err).Println()
}

func (ts *routerProvider) register(name string, fac ActorAdapterFactory, opts *ServiceOptions) (*entity, error) {
	if ts.isRunning {
		return nil, errors.New("the dcs service is running, please wait stopped to register") //分布式服务已经运行
	}
	m := meta{
		ServiceID: name,
		BLMode:    opts.mode,
		State:     statePreparing,
	}

	if opts.maxActor == 0 && opts.mode != BLModeHash {
		opts.SetMaxActorCnt(initPollActorNum)
	}
	s := newEntityServiceLocal(m, fac, opts)
	r := newRouter(m.ServiceID, opts.mode, s)
	ts.localService.Store(m.ServiceID, r)
	return s, nil
}

func (ts *routerProvider) start(rpcAddr string) (err error) {
	//循环本地Router,逐一启动服务,检查虚拟节点,初始化并注册
	ts.mu.Lock()
	defer ts.mu.Unlock()
	ts.localService.Range(func(key, value any) bool {
		r := value.(*router)
		ts.routerMap.Store(r.serviceName, r)
		if err = r.startLocalService(rpcAddr); err != nil { //初始化本地节点
			err = errors.Wrap(err, "init local service")
		}
		return err == nil
	})
	if err != nil {
		ts.routerMap.Range(func(key, value any) bool {
			r := value.(*router)
			r.stopLocalService()
			return true
		})
		ts.routerMap = sync.Map{}
	}
	ts.isRunning = err == nil
	return err
}

func (ts *routerProvider) stop() {
	ts.mu.Lock()
	defer ts.mu.Unlock()
	if !ts.isRunning {
		return
	}
	wgw := go_pool.NewGoWaitFunc(nil)
	ts.localService.Range(func(key, value any) bool {
		r := value.(*router)
		wgw.Run("stop local service", func(...any) {
			r.stopLocalService()
		})
		return true
	})
	wgw.Wait()
	ts.isRunning = false
	ts.routerMap.Range(func(key, value any) bool {
		r := value.(*router)
		r.wait()
		return true
	})
	ts.retryRunner.Wait()
	ts.routerMap = sync.Map{} //重置
}

func (ts *routerProvider) retry(msg *message.Message, err error) {
	ts.mu.Lock()
	defer ts.mu.Unlock()
	if !ts.isRunning {
		doMsgPostErr(msg, err)
		return
	}
	if ok, t := msg.AddRetryTimes(); ok {
		_ = ts.retryRunner.AsyncRun(func(args ...any) {
			m := args[0].(*message.Message)
			doTime := args[1].(time.Time)
			<-time.After(doTime.Sub(time.Now()))
			ts.post(m)
		}, msg, t)
	} else {
		doMsgPostErr(msg, err)
	}
}

func (ts *routerProvider) postUndo(msg *message.Message) {
	_ = ts.retryRunner.AsyncRun(func(args ...any) {
		runtime.Gosched()
		m := args[0].(*message.Message)
		ts.post(m)
	}, msg)
}

func (ts *routerProvider) post(msg *message.Message) {
	r, err := ts.takeRouter(msg.GetToActor().ServiceID())
	if err != nil {
		doMsgPostErr(msg, err)
		return
	}
	r.post(msg)
}

func (ts *routerProvider) broadcast(srvName string, msg *message.Message) bool {
	r, err := ts.takeRouter(srvName)
	if err != nil {
		doMsgPostErr(msg, err)
		return false
	}
	return r.broadcast(msg)
}

func (ts *routerProvider) watchService(serviceName string, mode BLMode) error {
	if _, ok := ts.routerMap.Load(serviceName); ok { //非本地服务,第一次获取此服务路由
		return nil
	}

	ts.mu.Lock()
	defer ts.mu.Unlock()
	if _, ok := ts.routerMap.Load(serviceName); ok {
		return nil
	}

	var r *router
	id, infos, _ := discov.GetServices(serviceName)

	fs := make([]service, 0, len(infos))
	for _, v := range infos {
		f, err2 := takeServiceWithJson(v, nil)
		if err2 != nil {
			log.GetLogger().Error("service with json").Str("js", v).Println()
			continue
		}
		fs = append(fs, f)
	}

	if len(fs) == 0 {
		r = newRouter(serviceName, mode, nil)

	} else {
		r = newRouter(serviceName, fs[0].getBLMode(), nil)
	}

	ts.routerMap.Store(serviceName, r)
	for _, f := range fs {
		r.registerService(f)
	}

	return r.watch(id)
}

func (ts *routerProvider) takeRouter(serviceName string) (*router, error) {
	if rCtx, ok := ts.routerMap.Load(serviceName); ok { //非本地服务,第一次获取此服务路由
		return rCtx.(*router), nil
	}
	ts.mu.Lock()
	defer ts.mu.Unlock()
	if rCtx, ok := ts.routerMap.Load(serviceName); ok {
		return rCtx.(*router), nil
	}
	var r *router
	id, infos, err := discov.GetServices(serviceName)
	if err != nil {
		return nil, errors.WithMessage(err, "take router from svc_disco")
	}

	if len(infos) == 0 {
		return nil, errors.New("take router not find service:" + serviceName)
	}

	fs := make([]service, 0, len(infos))
	for _, v := range infos {
		f, err2 := takeServiceWithJson(v, nil)
		if err2 != nil {
			log.GetLogger().Error("service with json").Str("js", v).Println()
			continue
		}
		fs = append(fs, f)
	}

	r = newRouter(serviceName, fs[0].getBLMode(), nil)
	ts.routerMap.Store(serviceName, r)
	for _, f := range fs {
		r.registerService(f)
	}
	if err = r.watch(id); err != nil {
		return nil, errors.WithMessage(err, "take router watch")
	}
	return r, nil
}
